package com.lagou.rpc.consumer.zk;

import com.alibaba.fastjson.JSONObject;
import com.lagou.rpc.consumer.client.RpcClient;
import lombok.Data;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * zookeeper客户端
 *
 * @Author: zhangzhiwei
 * @Date: 2022/2/14 22:00
 */
@Data
@Component
public class ConsumerZk{

    private CuratorFramework zkClient;

    // key = /provider/p0000000022    value = 127.0.0.1:8900
    private Map<String, RpcClient> map = new HashMap<>();

    public void connect() throws Exception {
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        zkClient = CuratorFrameworkFactory.builder()
                .connectString("192.168.229.131:2181")
                .sessionTimeoutMs(5000)
                .connectionTimeoutMs(3000)
                .retryPolicy(retryPolicy)
                .namespace("rpc")
                .build();
        zkClient.start();

        // 获取注册的服务端信息
        /*List<String> strings = zkClient.getChildren().forPath("/provider");
        if(strings != null && strings.size() != 0){
            for (String s : strings) {
                // 与netty服务端建立连接，并存储到集合中
                String ipPort = new String(zkClient.getData().forPath("/provider/" + s));
                System.out.println("==========从zk上获取服务端:" + ipPort + "==========");
                String[] split = ipPort.split(":");
                RpcClient rpcClient = new RpcClient(split[0], Integer.parseInt(split[1]));
                map.put(s, rpcClient);
            }
        }*/

        // 设置监听 并 获取注册的服务端信息
        PathChildrenCache cache = new PathChildrenCache(zkClient, "/provider", true);
        PathChildrenCacheListener listener = new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent event) throws Exception {
                ChildData data = event.getData();
                if(data == null){
                    return;
                }
                String path = data.getPath();
                String jsonContent = new String(data.getData());
                JSONObject jsonObject = JSONObject.parseObject(jsonContent);
                String ip = (String) jsonObject.get("ip");
                int port = (Integer) jsonObject.get("port");
                switch (event.getType()){
                    case INITIALIZED:
                        System.out.println("监听初始化完成");
                        break;
                    case CHILD_ADDED:
                        System.out.println("子节点增加:" + path + ",data=" + ip + ":" + port);
                        map.put(path, new RpcClient(ip, port));
                        System.out.println("与服务端" + ip + ":" + port + "建立连接成功");
                        break;
                    /*case CHILD_UPDATED:
                        System.out.println("子节点修改:" + path + ",data=" + ip + ":" + port);
                        map.put(path, new RpcClient(ip, port));
                        break;*/
                    case CHILD_REMOVED:
                        System.out.println("子节点删除:" + path + ",data=" + ip + ":" + port);
                        System.out.println("与服务端" + ip + ":" + port + "断开连接");
                        map.remove(path);
                        break;
                    default:
                        break;
                }
            }
        };
        cache.getListenable().addListener(listener);
        /*
         * StartMode.NORMAL                 异步初始化cache:启动时，异步初始化cache，完成后不会发出通知。不会触发PathChildrenCacheEvent.Type.INITIALIZED
         * StartMode.BUILD_INITIAL_CACHE    同步初始化cache:启动时，同步初始化cache，以及创建cache后，就从服务器拉取对应的数据。
         * StartMode.POST_INITIALIZED_EVENT 异步初始化cache，并触发完成事件:启动时，异步初始化cache，初始化完成触发。会触发PathChildrenCacheEvent.Type.INITIALIZED
         */
        cache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
    }

}
